package de.preissler.createview;

import java.util.Arrays;
import java.util.LinkedList;
import java.util.List;

import me.prettyprint.cassandra.serializers.DoubleSerializer;
import me.prettyprint.cassandra.serializers.LongSerializer;
import me.prettyprint.cassandra.serializers.StringSerializer;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.beans.HColumn;
import me.prettyprint.hector.api.beans.HSuperColumn;
import me.prettyprint.hector.api.beans.Row;
import me.prettyprint.hector.api.factory.HFactory;
import me.prettyprint.hector.api.mutation.Mutator;
import me.prettyprint.hector.api.query.RangeSlicesQuery;
import me.prettyprint.hector.api.query.SubSliceQuery;

/**
 * Erstellt einen RVT für Quotes by Change
 * @author Eric Preissler
 *
 */
public class QuotesByChange
{
	private static final String QUOTES = "Quotes";
	private static final String DAX = "DAX";
	private static final String CHANGE = "change";
	
	Keyspace keyspace;
	
	/**
	 * Konstruktur fuer Quotes by change
	 * @param keyspace
	 */
	public QuotesByChange(Keyspace keyspace)
	{
		this.keyspace = keyspace;
	}
	
	/**
	 * Erstellt den RVT
	 * @throws Exception
	 */
	public void createIndex() throws Exception
	{
		//alle verfuegbaren ISINS und Dates holen
		List<String> isins = getKeys(keyspace, QUOTES);
		List<String> dates = getKeys(keyspace, DAX);
		List<Exception> exceptions = new LinkedList<Exception>();
		int i = 1;
		
		for(String isin : isins)
		{
			MyUtil.trace("Working on " + isin + " (" + i + "/" + isins.size() + ")");
			for(String date : dates)
			{
				try
				{
					createRecord(isin, date, keyspace);
				}
				catch(Exception e)
				{
					exceptions.add(e);
				}
			}
			i++;
		}
		

		MyUtil.trace("Exceptions for QuotesByChange: " + exceptions.size());
	}
	
	/*
	 * Fuegt eine Zeile in den Index
	 */
	@SuppressWarnings("unchecked")
	private void createRecord(String isin, String date, Keyspace keyspace) throws Exception
	{
		SubSliceQuery<String, String, String, Double> query = HFactory.createSubSliceQuery(keyspace,
				StringSerializer.get(), StringSerializer.get(), StringSerializer.get(), DoubleSerializer.get());
		query.setColumnFamily(QUOTES);
		query.setKey(isin);
		query.setSuperColumn(date);
		query.setColumnNames(CHANGE);
		double change = query.execute().get().getColumnByName(CHANGE).getValue();
		long changeLong = (long)(change * 1000000.0);
		
		Mutator<String> mutator = HFactory.createMutator(keyspace, StringSerializer.get());
		HColumn<String, String> column = HFactory.createStringColumn(isin, "");
		HSuperColumn<Long, String, String> sc = HFactory.createSuperColumn(changeLong, Arrays.asList(column),
				LongSerializer.get(), StringSerializer.get(), StringSerializer.get());
		mutator.insert(date, Main.SCF_QUOTES_BY_CHANGE, sc);
	}
	
	/*
	 * Holt alle ISINs raus
	 */
	private List<String> getKeys(Keyspace keyspace, String nameOfColumnFamily)
	{
		List<List<String>> listsOfISINs = new LinkedList<List<String>>();
		
		String start = null;
		String end = null;
		boolean firstLoop = true;
		//Max. werden nur 100 Schluessel zurueck gegeben, deswegen diese Schleife
		while(true)
		{
			List<String> isins = new LinkedList<String>();
			
	         RangeSlicesQuery<String, String, String> rangeSlicesQuery = 
	        	 HFactory.createRangeSlicesQuery(keyspace, StringSerializer.get(), StringSerializer.get(), StringSerializer.get());
		     rangeSlicesQuery.setColumnFamily(nameOfColumnFamily);
		     rangeSlicesQuery.setReturnKeysOnly();
		     rangeSlicesQuery.setKeys(start, end);
		     List<Row<String, String, String>> result = rangeSlicesQuery.execute().get().getList();
		     
		     for(int i = 0; i < result.size(); i++)
		     {
		    	 if((!firstLoop) && (i == 0))
		    	 {
		    		 
		    	 }
		    	 else
		    		 isins.add(result.get(i).getKey());
		    		 
		     }
		     
		     if(isins.isEmpty())
		    	 break;
		     else
		     {
		    	 start = isins.get(isins.size()-1);
		    	 listsOfISINs.add(isins);
		    	 firstLoop = false;
		     }
		}
		
		List<String> isins = new LinkedList<String>();
		for(List<String> subList : listsOfISINs)
		{
			isins.addAll(subList);
		}
	     
		return isins;
	}
}
